/*
 * Author:   W.SH
 */
package com.wgg.crawler.scheduler;

import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.logging.Level;
import java.util.logging.Logger;

import com.wgg.crawler.filter.BloomFilter;

/**
 * 〈一句话功能简述〉<br> e
 * 〈功能详细描述〉
 *
 *  @author W.SH
 */
public class BloomDepthScheduler extends SchedulerMonitor implements Scheduler {

    private static final Logger LOG = Logger.getLogger(BloomDepthScheduler.class.getName());
    
    private final Object lock = new Object();
    
    private BlockingQueue<String> currentQueue = new LinkedBlockingDeque<String>();
    private BlockingQueue<String> nextQueue = new LinkedBlockingDeque<String>();
    private BloomFilter<String> bloomFilter = new BloomFilter<String>(0.0001, 10*1000*1000);
    
    @Override
    public String poll() throws InterruptedException {
        takeTaskCount.incrementAndGet();
        try {
            return currentQueue.poll();
        } catch (Exception e) {
            return null;
        }
    }

    @Override
    public void offer(List<String> urls) {
        synchronized(lock) {
            for(String url : urls) {
                if (null != url && !bloomFilter.contains(url)) {
                    putTaskCount.incrementAndGet();
                    nextQueue.offer(url);
                }
            }
        }
    }
        
    public void merge() {
        LOG.log(Level.INFO, "----> SchedulerQueue merge start <----");
        synchronized(lock) {
            currentQueue.clear();
            currentQueue.addAll(nextQueue);
            nextQueue.clear();
            depthCounter.incrementAndGet();
        }
        LOG.log(Level.INFO, "----> SchedulerQueue merge end <----");
    }
}
